/**
 * Description:
 * Class Name:ThreadPoolAIOServer
 * Date:2022/5/26 18:03
 *
 * @author LTisme
 * @version JAVA17
 */
package 疯狂Java讲义.第17章_网络编程._17_3_基于TCP协议的网络编程._17_3_8_使用AIO实现非阻塞通信;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolAIOServer {
    static final int PORT = 30000;
    static List<AsynchronousSocketChannel> channelList = new ArrayList<>();

    public void startListen() throws InterruptedException, Exception{
        // 创建一个线程池
        ExecutorService executor = Executors.newFixedThreadPool(20);
        // 以指定线程池来创建一个AsynchronousChannelGroup
        AsynchronousChannelGroup channelGroup = AsynchronousChannelGroup.withThreadPool(executor);
        // 以指定线程池来创建一个AsynchronousServerSocketChannel，3个步骤open()、bind()、accept()
        AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open(channelGroup);
        serverSocketChannel.bind(new InetSocketAddress(PORT));
        // 使用CompletionHandler接受来自客户端的连接请求
        serverSocketChannel.accept(null, new AcceptHandler(serverSocketChannel));           // ①
        Thread.sleep(100000);
    }

    public static void main(String[] args) throws Exception{
        var server = new ThreadPoolAIOServer();
        server.startListen();
    }
}

// 实现自己的CompletionHandler类
class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object>{
    private AsynchronousServerSocketChannel serverSocketChannel;
    public AcceptHandler(AsynchronousServerSocketChannel serverSocketChannel){
        this.serverSocketChannel = serverSocketChannel;
    }
    // 定义一个ByteBuffer准备来读取数据
    ByteBuffer buffer = ByteBuffer.allocate(1024);

    // 当实际 IO 操作完成时触发该方法
    @Override
    public void completed(final AsynchronousSocketChannel sc, Object attachment) {
        // 记录新连接进来的Channel
        ThreadPoolAIOServer.channelList.add(sc);
        // 准备接收客户端的下一次连接
        serverSocketChannel.accept(null, this);
        sc.read(buffer, null,
                new CompletionHandler<Integer, Object>()                                                // ②
        {
            @Override
            public void completed(Integer result, Object attachment) {
                buffer.flip();
                String content = StandardCharsets.UTF_8.decode(buffer).toString();
                // 遍历每个Channel，将收到的信息写入各Channel中
                for (AsynchronousSocketChannel c :
                        ThreadPoolAIOServer.channelList) {
                    try {
                        c.write(ByteBuffer.wrap(content.getBytes(StandardCharsets.UTF_8))).get();
                    }
                    catch (Exception exception){
                        exception.printStackTrace();
                    }
                }
                buffer.clear();
                // 读取下一次数据
                sc.read(buffer, null, this);
            }

            @Override
            public void failed(Throwable exc, Object attachment) {
                System.out.println("读取数据失败：" + exc);
                // 从该Channel中读取数据失败，就将该Channel删除
                ThreadPoolAIOServer.channelList.remove(sc);
            }
        });
    }

    @Override
    public void failed(Throwable exc, Object attachment) {
        System.out.println("连接失败：" + exc);
    }
}
